package realtime.dws;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.formats.json.JsonDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import realtime.bean.OdsUserBehavior;
import realtime.constant.Constant;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;

/**
 * @author shenyucheng
 * @version 1.0
 * @description: TODO
 * @date 2025/3/14 21:28
 */
public class DwdGong {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Kafka 配置
        KafkaSource<String> data = getGroupId();
        DataStreamSource<String> source = env.fromSource(data, WatermarkStrategy.noWatermarks(), "kafka_source");

        DataStream<OdsUserBehavior> odsStream = source.map(new MapFunction<String, OdsUserBehavior>() {
            @Override
            public OdsUserBehavior map(String value) throws Exception {
                return JSON.parseObject(value, OdsUserBehavior.class);
            }
        });

        // DWD 层处理：添加是否首次访问标识，清洗数据
        DataStream<OdsUserBehavior> dwdStream = odsStream.keyBy(OdsUserBehavior::getUser_id)
                .process(new KeyedProcessFunction<String, OdsUserBehavior, OdsUserBehavior>() {
                    private ValueState<Boolean> isFirstVisitState;

                    @Override
                    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
                        isFirstVisitState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("isFirstVisit", Boolean.class)
                        );
                    }

                    @Override
                    public void processElement(OdsUserBehavior value, Context ctx, Collector<OdsUserBehavior> out) throws Exception {
                        Boolean isFirstVisit = isFirstVisitState.value();
                        if (isFirstVisit == null) {
                            isFirstVisit = true;
                            isFirstVisitState.update(true);
                        }
                        value.setIs_first_visit(isFirstVisit); // 假设 OdsUserBehavior 类添加了 is_first_visit 字段
                        out.collect(value);
                    }
                });
        dwdStream.print();
        // DWS 层处理：计算新访未成交和新访成交指标
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.createTemporaryView("dwd_user_behavior", dwdStream);
        // 计算新访未成交指标
        String newVisitNoPurchaseSql = "SELECT \n" +
                "    COUNT(DISTINCT user_id) AS new_visit_no_purchase_count \n" +
                "FROM dwd_user_behavior \n" +
                "WHERE is_first_visit = true \n" +
                "  AND behavior_type IN ('访问', '互动') \n" +
                "  AND user_id NOT IN ( \n" +
                "      SELECT DISTINCT user_id \n" +
                "      FROM dwd_user_behavior \n" +
                "      WHERE behavior_type = '支付' \n" +
                "      AND TO_DATE(CAST(behavior_time AS STRING), 'yyyy-MM-dd HH:mm:ss') = TO_DATE(CAST(CURRENT_TIMESTAMP AS STRING), 'yyyy-MM-dd HH:mm:ss') \n" +
                "  )";
        Table newVisitNoPurchaseTable = tEnv.sqlQuery(newVisitNoPurchaseSql);
//        newVisitNoPurchaseTable.execute().print();

       // 计算新访成交指标
        String newVisitPurchaseSql = "SELECT " +
                "COUNT(DISTINCT user_id) AS new_visit_purchase_count " +
                "FROM dwd_user_behavior " +
                "WHERE is_first_visit = true " +
                "AND behavior_type IN ('访问', '互动') " +
                "AND user_id IN ( " +
                "    SELECT DISTINCT user_id " +
                "    FROM dwd_user_behavior " +
                "    WHERE behavior_type = '支付' " +
                "    AND TO_DATE(CAST(behavior_time AS STRING)) = TO_DATE(CAST(CURRENT_TIMESTAMP AS STRING)) " +
                ")";
        Table newVisitPurchaseTable = tEnv.sqlQuery(newVisitPurchaseSql);
//        DataStream<Long> newVisitPurchaseStream = tEnv.toDataStream(newVisitPurchaseTable, Long.class);
        newVisitPurchaseTable.execute().print();



        env.execute();
    }

    private static KafkaSource<String> getGroupId() {
        return KafkaSource.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                .setGroupId("groupId")
                .setTopics("topic-gong")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] message) throws IOException {
                        if (message != null) {
                            return new String(message, StandardCharsets.UTF_8);
                        }
                        return null;
                    }

                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return Types.STRING;
                    }
                })
                .build();
    }
}
